/*
 * Copyright 2022 LMAX Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.lmax.disruptor;

import java.util.concurrent.atomic.AtomicInteger;

import static com.lmax.disruptor.RewindAction.REWIND;
import static java.lang.Math.min;


/**
 * Convenience class for handling the batching semantics of consuming entries from a {@link RingBuffer}
 * and delegating the available events to an {@link EventHandler}.
 *
 * @param <T> event implementation storing the data for sharing during exchange or parallel coordination of an event.
 */
public final class BatchEventProcessor<T>
        implements EventProcessor
{
    private static final int IDLE = 0;
    private static final int HALTED = IDLE + 1;
    private static final int RUNNING = HALTED + 1;

    private final AtomicInteger running = new AtomicInteger(IDLE); // 是否启动的标识
    private ExceptionHandler<? super T> exceptionHandler; // 绑定的 ExceptionHandler
    private final DataProvider<T> dataProvider; // 所消费的RingBuffer
    private final SequenceBarrier sequenceBarrier; // 依赖的SequenceBarrier（序列屏障）
    private final EventHandlerBase<? super T> eventHandler; // 绑定的 EventHandler
    private final int batchLimitOffset;// 默认 Integer.MAX_VALUE-1

    // 消费者想要获取到 RingBuffer 中的元素，就需要从 Sequence 中取得可用的序列号，否则就会执行等待策略。
    // EventHandler 最终封装为 BatchEventProcessor，每个 BatchEventProcessor 在执行 EventHandler 相应逻辑之前都会先获取可用的序列号，
    // 因为每个 BatchEventProcessor 独立维护了一个 Sequence 对象，所以每个事件都会被所有的消费者处理一遍。
    private final Sequence sequence = new Sequence(Sequencer.INITIAL_CURSOR_VALUE); // 消费的序列号
    private final RewindHandler rewindHandler;
    private int retriesAttempted = 0;

    BatchEventProcessor(
            final DataProvider<T> dataProvider, // 所消费的RingBuffer
            final SequenceBarrier sequenceBarrier, // 依赖的SequenceBarrier（序列屏障）
            final EventHandlerBase<? super T> eventHandler, // 绑定的 EventHandler
            final int maxBatchSize, // 默认 Integer.MAX_VALUE-1
            final BatchRewindStrategy batchRewindStrategy
    )
    {
        this.dataProvider = dataProvider;
        this.sequenceBarrier = sequenceBarrier;
        this.eventHandler = eventHandler;

        if (maxBatchSize < 1)
        {
            throw new IllegalArgumentException("maxBatchSize must be greater than 0");
        }
        this.batchLimitOffset = maxBatchSize - 1;

        this.rewindHandler = eventHandler instanceof RewindableEventHandler
                ? new TryRewindHandler(batchRewindStrategy)
                : new NoRewindHandler();
    }

    @Override
    public Sequence getSequence()
    {
        return sequence;
    }

    @Override
    public void halt()
    {
        // 指示此 EventProcessor 在下一次干净中断时完成消耗时应停止。
        // 它将调用 SequenceBarrier.alert() 以通知线程检查状态。
        running.set(HALTED);
        sequenceBarrier.alert();
    }

    @Override
    public boolean isRunning()
    {
        return running.get() != IDLE;
    }

    /**
     * Set a new {@link ExceptionHandler} for handling exceptions propagated out of the {@link BatchEventProcessor}.
     *
     * @param exceptionHandler to replace the existing exceptionHandler.
     */
    public void setExceptionHandler(final ExceptionHandler<? super T> exceptionHandler)
    {
        if (null == exceptionHandler)
        {
            throw new NullPointerException();
        }

        this.exceptionHandler = exceptionHandler;
    }

    /**
     * It is ok to have another thread rerun this method after a halt().
     * 在 halt（） 之后让另一个线程重新运行此方法是可以的
     *
     * @throws IllegalStateException if this object instance is already running in a thread
     */
    @Override
    public void run()
    {
        // 通过CAS的方式将 running由IDLE改成RUNNING
        // 如果CAS成功返回原来running的值
        int witnessValue = running.compareAndExchange(IDLE, RUNNING);
        if (witnessValue == IDLE) // Successful CAS
        {
            // 此时running==RUNNING

            // 清除sequenceBarrier警报状态
            sequenceBarrier.clearAlert();

            // EventProcessor启动成功 通知 EventHandler
            notifyStart();
            try
            {
                if (running.get() == RUNNING)
                {
                    // 处理事件
                    processEvents();
                }
            }
            finally
            {
                // 在此处理器关闭之前立即通知 EventHandler
                notifyShutdown();
                running.set(IDLE);
            }
        }
        else
        {
            if (witnessValue == RUNNING)
            {
                // 重复启动了
                throw new IllegalStateException("Thread is already running");
            }
            else
            {
                // 【？？？】
                // notifyStart();
                // notifyShutdown();
                earlyExit();
            }
        }
    }

    private void processEvents()
    {
        // 消费者想要获取到 RingBuffer 中的元素，就需要从 Sequence 中取得可用的序列号，否则就会执行等待策略。
        // EventHandler 最终封装为 BatchEventProcessor，每个 BatchEventProcessor 在执行 EventHandler 相应逻辑之前都会先获取可用的序列号，
        // 因为每个 BatchEventProcessor 独立维护了一个 Sequence 对象，所以每个事件都会被所有的消费者处理一遍。

        T event = null;
        long nextSequence = sequence.get() + 1L;

        // 一次循环一次批处理
        while (true)
        {
            // 此次批处理的起始Sequence
            final long startOfBatchSequence = nextSequence;
            try
            {
                try
                {
                    // 等待可消费的Sequence，在sequenceBarrier.waitFor()内部会执行waitStrategy.waitFor()方法实现生产者和消费者之间的通信。
                    // 通过SequenceBarrier隔离各个消费者的消费空间
                    final long availableSequence = sequenceBarrier.waitFor(nextSequence);

                    // 此次批处理的末尾Sequence
                    final long endOfBatchSequence = min(nextSequence + batchLimitOffset, availableSequence);

                    if (nextSequence <= endOfBatchSequence)
                    {
                        // 执行处理这一批Event之前的逻辑 默认空实现 用户可以自行选择是否实现
                        eventHandler.onBatchStart(endOfBatchSequence - nextSequence + 1, availableSequence - nextSequence + 1);
                    }

                    // 开始进行Event数据消费
                    while (nextSequence <= endOfBatchSequence)
                    {
                        // 获取该序列下的Event
                        event = dataProvider.get(nextSequence);
                        // 执行处理Event逻辑 用户必须根据需求实现
                        // 也可以根据是否是该批次最后的一个事件来做出相应的处理逻辑
                        eventHandler.onEvent(event, nextSequence, nextSequence == endOfBatchSequence);
                        nextSequence++;
                    }

                    retriesAttempted = 0;

                    // 设置消费者当前的消费进度至sequence
                    sequence.set(endOfBatchSequence);
                }
                catch (final RewindableException e)
                {
                    nextSequence = rewindHandler.attemptRewindGetNextSequence(e, startOfBatchSequence);
                }
            }
            catch (final TimeoutException e)
            {
                notifyTimeout(sequence.get());
            }
            catch (final AlertException ex)
            {
                if (running.get() != RUNNING)
                {
                    break;
                }
            }
            catch (final Throwable ex)
            {
                // 执行onEvent抛出的业务异常
                handleEventException(ex, nextSequence, event);
                // 设置消费者当前的消费进度至sequence
                sequence.set(nextSequence);
                nextSequence++;
            }
        }
    }

    private void earlyExit()
    {
        notifyStart();
        notifyShutdown();
    }

    private void notifyTimeout(final long availableSequence)
    {
        try
        {
            // 【？？？】默认为空实现
            eventHandler.onTimeout(availableSequence);
        }
        catch (Throwable e)
        {
            handleEventException(e, availableSequence, null);
        }
    }

    /**
     * Notifies the EventHandler when this processor is starting up.
     */
    private void notifyStart()
    {
        try
        {
            // 【？？？】默认为空实现
            eventHandler.onStart();
        }
        catch (final Throwable ex)
        {
            handleOnStartException(ex);
        }
    }

    /**
     * Notifies the EventHandler immediately prior to this processor shutting down.
     * 在此处理器关闭之前立即通知 EventHandler
     */
    private void notifyShutdown()
    {
        try
        {
            // 【？？？】默认为空实现
            eventHandler.onShutdown();
        }
        catch (final Throwable ex)
        {
            handleOnShutdownException(ex);
        }
    }

    /**
     * Delegate to {@link ExceptionHandler#handleEventException(Throwable, long, Object)} on the delegate or
     * the default {@link ExceptionHandler} if one has not been configured.
     */
    private void handleEventException(final Throwable ex, final long sequence, final T event)
    {
        getExceptionHandler().handleEventException(ex, sequence, event);
    }

    /**
     * Delegate to {@link ExceptionHandler#handleOnStartException(Throwable)} on the delegate or
     * the default {@link ExceptionHandler} if one has not been configured.
     */
    private void handleOnStartException(final Throwable ex)
    {
        getExceptionHandler().handleOnStartException(ex);
    }

    /**
     * Delegate to {@link ExceptionHandler#handleOnShutdownException(Throwable)} on the delegate or
     * the default {@link ExceptionHandler} if one has not been configured.
     */
    private void handleOnShutdownException(final Throwable ex)
    {
        getExceptionHandler().handleOnShutdownException(ex);
    }

    private ExceptionHandler<? super T> getExceptionHandler()
    {
        ExceptionHandler<? super T> handler = exceptionHandler;
        return handler == null ? ExceptionHandlers.defaultHandler() : handler;
    }

    private class TryRewindHandler implements RewindHandler
    {
        private final BatchRewindStrategy batchRewindStrategy;

        TryRewindHandler(final BatchRewindStrategy batchRewindStrategy)
        {
            this.batchRewindStrategy = batchRewindStrategy;
        }

        @Override
        public long attemptRewindGetNextSequence(final RewindableException e, final long startOfBatchSequence) throws RewindableException
        {
            if (batchRewindStrategy.handleRewindException(e, ++retriesAttempted) == REWIND)
            {
                return startOfBatchSequence;
            }
            else
            {
                retriesAttempted = 0;
                throw e;
            }
        }
    }

    private static class NoRewindHandler implements RewindHandler
    {
        @Override
        public long attemptRewindGetNextSequence(final RewindableException e, final long startOfBatchSequence)
        {
            throw new UnsupportedOperationException("Rewindable Exception thrown from a non-rewindable event handler", e);
        }
    }
}